﻿/*
Copyright (C) 2009  Tomasz Chrzanowski

This program is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the GNU General Public License
along with this program.  If not, see <http://www.gnu.org/licenses/>
*/

using System;
using System.Collections.Generic;
using System.Data;
using System.Data.SqlClient;
using System.Text;

namespace Com.TomCdc.ChangeDataCapture
{
    /// <summary>
    /// Provides method to create and populate properties of <see cref="SourceTable"/> object.
    /// <remarks>
    /// Before using, the object has to be initialized by calling <see cref="InitializeFactory"/>.
    /// </remarks>
    /// <example>
    /// SourceSchemaFactory ssf = new SourceSchemaFactory(connString);
    /// ssf.InitializeFactory(); // might be time consuming
    /// DataTable cust = ssf.CreateSourceTable("Customers"); // the Customer table must exists
    /// </example>
    /// </summary>
    public class SourceSchemaFactory
    {
        /// <summary>
        /// Sql connection string to source database.
        /// </summary>
        private string _sourceDbConnString;

        /// <summary>
        /// Sql connection string to storage database.
        /// </summary>
        private string _storeDbConnString;

        /// <summary>
        /// Stores the result of SQL_SELECT_TABLE_SCHEMA query execution.
        /// </summary>
        private DataTable _srcTableCol;

        /// <summary>
        /// Stores the result of SQL_SELECT_TRACKED_TABLES query execution.
        /// </summary>
        private DataTable _srcTrackedTable;

        /// <summary>
        /// Stores the result of SQL_SELECT_PRIMARY_UNIQUE_KEY query execution.
        /// </summary>
        private DataTable _srcTablePrimKeyUniqueCol;

        /// <summary>
        /// Stores the result of SQL_SELECT_STORE_COLUMN_ORDINAL query execution.
        /// </summary>
        private DataTable _storeTableColId;

        /// <summary>
        /// Set to true if the object was initialized.
        /// </summary>
        private bool _isInitialized;

        /// <summary>
        /// The trigger prefix.
        /// </summary>
        private const string TRIGGER_PREFIX = "trCDC__";

        /// <summary>
        /// The place holder of the trigger prefix in a template file.
        /// </summary>
        private const string PLACE_HOLDER_TRIGGER_PREFIX = "{triggerPrefix}";

        /// <summary>
        /// The version of sql server.
        /// </summary>
        private SqlServerVersion _serverVersion;

        /// <summary>
        /// The sql statement to retreive list of tables, columns and its types.
        /// </summary>
        private const string SQL_SELECT_TABLE_SCHEMA = @"
            SELECT 
                T.TABLE_SCHEMA
                ,C.TABLE_NAME
                ,C.COLUMN_NAME
                ,ISNULL(COLUMNPROPERTY(OBJECT_ID(T.TABLE_SCHEMA + '.' + T.TABLE_NAME), C.COLUMN_NAME, 'ColumnID'), C.ORDINAL_POSITION) ORDINAL_POSITION
                ,C.DATA_TYPE
                ,C.CHARACTER_MAXIMUM_LENGTH
            FROM 
                INFORMATION_SCHEMA.TABLES T 
                INNER JOIN INFORMATION_SCHEMA.COLUMNS C ON T.TABLE_NAME=C.TABLE_NAME
            WHERE 
                T.TABLE_TYPE='BASE TABLE' AND C.TABLE_NAME NOT LIKE '#%'
            ORDER BY 
                T.TABLE_SCHEMA
                ,C.TABLE_NAME
                ,C.ORDINAL_POSITION";

        /// <summary>
        /// The sql statement to select currently tracked tables and its triggers text.
        /// </summary>
        private const string SQL_SELECT_TRACKED_TABLES_V9 = @"
            SELECT 
                SYSS.name TABLE_SCHEMA,
                SYSO2.name TABLE_NAME,
                SYSC.text TRIGGER_TEXT
            FROM 
                sys.objects SYSO1 INNER JOIN syscomments SYSC ON SYSO1.object_id=SYSC.ID
                INNER JOIN sys.objects SYSO2 ON SYSO2.object_id=SYSO1.parent_object_id
                INNER JOIN sys.schemas SYSS ON SYSO2.schema_id = syss.schema_id
            WHERE 
                SYSO1.type='TR' and SYSO1.name like '{triggerPrefix}%'
            ORDER BY 
                SYSS.name,
                SYSO2.name, 
                SYSC.colid";

        /// <summary>
        /// The sql statement to select currently tracked tables and its triggers text.
        /// </summary>
        private const string SQL_SELECT_TRACKED_TABLES_V8 = @"
            SELECT 
                SYSU.NAME TABLE_SCHEMA,
                SYSO2.name TABLE_NAME,
                SYSC.text TRIGGER_TEXT
            FROM 
                sysobjects SYSO1 INNER JOIN syscomments SYSC ON SYSO1.ID=SYSC.ID
                INNER JOIN sysobjects SYSO2 ON SYSO2.ID=SYSO1.parent_obj
                INNER JOIN sysusers SYSU ON SYSU.UID=SYSO2.UID
            WHERE 
                SYSO1.type='TR' and SYSO1.name like '{triggerPrefix}%'
            ORDER BY
                SYSU.NAME,
                SYSO2.name, 
                SYSC.colid";

        // TODO: include in the unique columns result unique indexes containing on column, remember to order result
        /// <summary>
        /// The sql statement to select primary keys and unique constaints of source database tables.
        /// </summary>
        private const string SQL_SELECT_PRIMARY_UNIQUE_KEY = @"
            SELECT
                A.TABLE_SCHEMA
                ,A.TABLE_NAME
                ,B.COLUMN_NAME
                ,A.CONSTRAINT_TYPE
            FROM
                (SELECT 
                    T.TABLE_SCHEMA
                    ,T.TABLE_NAME
                    ,TC.CONSTRAINT_NAME
                    ,TC.CONSTRAINT_TYPE
                FROM 
                    INFORMATION_SCHEMA.TABLES T
                    INNER JOIN INFORMATION_SCHEMA.TABLE_CONSTRAINTS TC ON T.TABLE_NAME=TC.TABLE_NAME
                    INNER JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE CCU ON TC.CONSTRAINT_NAME=CCU.CONSTRAINT_NAME
                WHERE 
                    TC.CONSTRAINT_TYPE IN ('PRIMARY KEY','UNIQUE')
                    AND T.TABLE_TYPE='BASE TABLE'
                GROUP BY
                    T.TABLE_SCHEMA, T.TABLE_NAME, TC.CONSTRAINT_NAME, TC.CONSTRAINT_TYPE
                HAVING
                    COUNT(*) = 1
                ) A INNER JOIN INFORMATION_SCHEMA.CONSTRAINT_COLUMN_USAGE B ON A.CONSTRAINT_NAME=B.CONSTRAINT_NAME
            ORDER BY 
                A.TABLE_NAME
                ,A.CONSTRAINT_TYPE";

        /// <summary>
        /// The sql statement to select store database table names and column names and id.
        /// </summary>
        private const string SQL_SELECT_STORE_COLUMN_ORDINAL = @"
            SELECT
                ST.Name TABLE_IDENTIFIER, SC.Name COLUMN_NAME, SC.OrdinalPosition ORDINAL_POSITION
            FROM
                SourceTable ST INNER JOIN SourceColumn SC ON ST.ID=SC.SourceTableId
            ORDER BY
                ST.ID";

        /// <summary>
        /// Gets a value indicating whether the object was initialized.
        /// </summary>
        public bool IsInitialized
        {
            get
            {
                return _isInitialized;
            }
        }

        /// <summary>
        /// Initializes new instance of <see cref="SourceSchemaFactory"/> class. Before using object, call 
        /// <see cref="InitializeFactory"/> method.
        /// </summary>
        /// <param name="sourceDbConnString">The connection string to source (tracked) database.</param>
        public SourceSchemaFactory(string sourceDbConnString)
        {
            InitPrivateMembers(sourceDbConnString);
        }

        /// <summary>
        /// Initializes new instance of <see cref="SourceSchemaFactory"/> class. Before using object, call 
        /// <see cref="InitializeFactory"/> method.
        /// </summary>
        /// <param name="sourceDbConnString">The connection string to source (tracked) database.</param>
        /// <param name="storeDbConnString">The connection string to storage database.</param>
        public SourceSchemaFactory(string sourceDbConnString, string storeDbConnString)
        {
            if (string.IsNullOrEmpty(storeDbConnString))
            {
                throw new ArgumentException("The storeDbConnString argument must not be empty or null.");
            }

            InitPrivateMembers(sourceDbConnString);

            _storeDbConnString = storeDbConnString;

            _storeTableColId = new DataTable("StoreTableColumnsId");
            _storeTableColId.CaseSensitive = false;
        }

        private void InitPrivateMembers(string sourceDbConnString)
        {
            if (string.IsNullOrEmpty(sourceDbConnString))
            {
                throw new ArgumentException("The sourceDbConnString argument must not be empty or null.");
            }

            _sourceDbConnString = sourceDbConnString;

            // prepares DataTable objects
            _srcTableCol = new DataTable("SourceTableColumn");
            _srcTableCol.CaseSensitive = false;

            _srcTrackedTable = new DataTable("SourceTrckedTable");
            _srcTrackedTable.CaseSensitive = false;

            _srcTablePrimKeyUniqueCol = new DataTable("SourceTablePrimKey");
            _srcTablePrimKeyUniqueCol.CaseSensitive = false;

            // marks object as not initialized, call InitializeFactory to initialize
            _isInitialized = false;
        }

        /// <summary>
        /// Initializes the <see cref="SourceSchemaFactory"/> object. Before using <code>SourceSchemaFactory</code>
        /// object, this mathod has to be called.
        /// </summary>
        public void InitializeFactory()
        {
            _serverVersion = CommonHelper.GetServerVersion(_sourceDbConnString);

            // creates scope for sql connection
            using (SqlConnection sqlConn = new SqlConnection(_sourceDbConnString))
            using (SqlCommand sqlCmd = new SqlCommand(SQL_SELECT_TABLE_SCHEMA, sqlConn))
            using (SqlDataAdapter sqlAdapter = new SqlDataAdapter(sqlCmd))
            {
                try
                {
                    // opens sql connection and in sequence populated DataTables
                    sqlConn.Open();

                    sqlAdapter.Fill(_srcTableCol);

                    string sqlTrackedTbl;
                    switch (_serverVersion)
                    {
                        case SqlServerVersion.Sql2000: sqlTrackedTbl = SQL_SELECT_TRACKED_TABLES_V8;
                            break;
                        default: sqlTrackedTbl = SQL_SELECT_TRACKED_TABLES_V9;
                            break;
                    }

                    sqlCmd.CommandText =
                        sqlTrackedTbl.Replace(PLACE_HOLDER_TRIGGER_PREFIX, TRIGGER_PREFIX);

                    sqlAdapter.Fill(_srcTrackedTable);

                    sqlCmd.CommandText = SQL_SELECT_PRIMARY_UNIQUE_KEY;

                    sqlAdapter.Fill(_srcTablePrimKeyUniqueCol);
                }
                catch (SqlException ex)
                {
                    throw new CdcSchemaInstallerException("The sql error ocurred while retrieving schema" +
                        " information from the source database.", ex);
                }
            }

            if (!string.IsNullOrEmpty(_storeDbConnString))
            {
                // creates scope for sql connection, sql command adn adapter
                using (SqlConnection sqlConn = new SqlConnection(_storeDbConnString))
                using (SqlCommand sqlCmd = new SqlCommand(SQL_SELECT_STORE_COLUMN_ORDINAL, sqlConn))
                using (SqlDataAdapter sqlAdapter = new SqlDataAdapter(sqlCmd))
                {
                    try
                    {
                        // opens sql connection and populates DataTable
                        sqlConn.Open();

                        sqlAdapter.Fill(_storeTableColId);
                    }
                    catch (SqlException ex)
                    {
                        throw new CdcSchemaInstallerException("The sql error ocurred while retrieving schema" +
                            " information from the store database.", ex);
                    }
                }
            }

            // marks the object as initialized
            _isInitialized = true;
        }

        /// <summary>
        /// Creates <see cref="SourceTable"/> object, and populates it with columns and all properties.
        /// </summary>
        /// <param name="tableName">The name of a table the <see cref="SourceTable"/> object will be 
        /// created for.</param>
        /// <param name="initStoreSrcColId">Determines if <see cref="SourceColumn"/> object property 
        /// <see cref="SourceColumn.StoringColumnId"/> will be initialized with storage database source column id.
        /// </param>
        /// <returns>The <see cref="SourceTable"/> object.</returns>
        public SourceTable CreateSourceTable(string tableSchema, string tableName, bool initStoreSrcColId)
        {
            // checks if factory initialized
            if (!_isInitialized)
            {
                throw new CdcSchemaInstallerException("The SourceSchemaFactory not initialized, call" +
                    " InitializeFactory method first.");
            }

            // validates input argument
            if (String.IsNullOrEmpty(tableSchema))
            {
                throw new ArgumentException("The tableSchema argument must not be empty or null.");
            }

            // validates input argument
            if (String.IsNullOrEmpty(tableName))
            {
                throw new ArgumentException("The tableName argument must not be empty or null.");
            }

            // prepares where clause for filtering data in DataTable
            string whereClause = string.Format("TABLE_SCHEMA = '{0}' AND TABLE_NAME = '{1}'",
                tableSchema.Replace("'", "''"), tableName.Replace("'", "''"));

            // contains list of column names and ids of source columns registered in the store database
            Dictionary<string, int> storeSrcColNameToId = null;

            // determines if store db columns id ready to use
            bool storeSrcColIdsReady = false;

            // checks if store col id init required
            if (initStoreSrcColId)
            {
                // checks if object constructed to support request
                if (string.IsNullOrEmpty(_storeDbConnString))
                {
                    throw new InvalidOperationException("Object not constracted to support true value of " +
                        "initStoreColId method argument.");
                }

                // slelect only rows related to requested table name
                DataRow[] storeColumns = _storeTableColId.Select(string.Format("TABLE_IDENTIFIER = '{0}{1}{2}'",
                    tableSchema, CommonHelper.SCHEMA_TABLE_SEPARATOR, tableName));

                // init dictionery with required capacity
                storeSrcColNameToId = new Dictionary<string, int>(storeColumns.Length);

                // populates dictionary
                foreach (DataRow row in storeColumns)
                {
                    storeSrcColNameToId[(string)row["COLUMN_NAME"]] = Convert.ToInt32(row["ORDINAL_POSITION"]);
                }

                // mark as ready to use
                storeSrcColIdsReady = true;
            }

            // gets tableName relevant triggers from the _srcTrackedTable
            DataRow[] tableTrigger = _srcTrackedTable.Select(whereClause);

            // concatenetes parts of the trigger into one string
            StringBuilder triggerTxtTmp = new StringBuilder();
            foreach (DataRow row in tableTrigger)
            {
                triggerTxtTmp.Append((string)row["TRIGGER_TEXT"]);
            }
            string triggerText = triggerTxtTmp.ToString();

            // searches for tokens which encompas columns list
            int colListStartTokenIdx = triggerText.IndexOf(CommonHelper.TRIGGER_COL_LIST_TOKEN);
            int colListEndTokenIdx = triggerText.IndexOf(CommonHelper.TRIGGER_COL_LIST_TOKEN, colListStartTokenIdx + 1);

            // prepares list of columns currently tracked 
            CaseInsensitiveList trackedColumns = new CaseInsensitiveList();
            if (colListStartTokenIdx > 0 && colListStartTokenIdx < colListEndTokenIdx)
            {
                // calculates start and end columns possition
                int colsStartIdx = colListStartTokenIdx + CommonHelper.TRIGGER_COL_LIST_TOKEN.Length;
                int colsLen = colListEndTokenIdx - colsStartIdx;

                // extracts columns list
                string cols = triggerText.Substring(colsStartIdx, colsLen);

                // converts comma separated list into collection
                trackedColumns.AddRange(cols.Split(new char[] { ',' }, StringSplitOptions.RemoveEmptyEntries));

                // removes squer brackets from begining and end
                for (int colIdx = 0; colIdx < trackedColumns.Count; colIdx++)
                {
                    trackedColumns[colIdx] = trackedColumns[colIdx].Substring(1, trackedColumns[colIdx].Length - 2);
                }
            }

            // gets tableName relevant table columns from the _srcTableCol
            DataRow[] tableColumns = _srcTableCol.Select(whereClause);

            // checks if the table name given in method parameter exists in database
            if (tableColumns.Length == 0)
            {
                throw new CdcSchemaInstallerException("The tableName argument contains not existing database table.");
            }

            // creates return object, and sets its name and tracking state (trigger exists means is tracked)
            SourceTable resSrcTable = new SourceTable(tableSchema, tableName, tableTrigger.Length > 0 ? true : false);

            // goes through all columns in the table and creates column objects for each of them
            foreach (DataRow row in tableColumns)
            {
                string colName = (string)row["COLUMN_NAME"];
                SourceColumn col = new SourceColumn(
                    colName,
                    (string)row["DATA_TYPE"],
                    // sql2005 returns int32, sql2000 int16, Convert.ToInt32 handles both cases
                    Convert.ToInt32(row["ORDINAL_POSITION"]));

                // checks if column tracked
                col.IsTracked = trackedColumns.Contains(col.ColumnName);

                if (!row.IsNull("CHARACTER_MAXIMUM_LENGTH") && (int)row["CHARACTER_MAXIMUM_LENGTH"] > 0)
                {
                    col.MaxLength = (int)row["CHARACTER_MAXIMUM_LENGTH"];
                }

                if (storeSrcColIdsReady)
                {
                    // assign only if column exists if store db dictionary
                    // this loop process all the track (source) database table columns, not all of them
                    // are being tracked, so dictionary entry doesn't exist
                    if (storeSrcColNameToId.ContainsKey(colName))
                    {
                        col.StoreSourceColumnOrdinal = storeSrcColNameToId[colName];
                    }
                }

                // adds created column into columns collection of the source table object
                resSrcTable.AddColumn(col);
            }

            // gets tableName relevant table primary key or unique constraint from the _srcTablePrimKeyUniqueCol 
            DataRow[] primaryKeys = _srcTablePrimKeyUniqueCol.Select(whereClause);

            // find first applicable PK/UNIQUE, if found, set primary key of the source table object
            foreach (DataRow primKeyRow in primaryKeys)
            {
                if (CommonHelper.AllowedPrimaryKeyTypes.Contains(
                    resSrcTable.Columns[(string)primKeyRow["COLUMN_NAME"]].DataType))
                {
                    resSrcTable.SetPrimaryKey((string)primKeyRow["COLUMN_NAME"]);
                    break;
                }
            }

            // identifies columns included in current capture trigger which do not exist in the tracked column
            // e.g. has been deleted since the trigger had been created
            foreach (string trackedCol in trackedColumns)
            {
                if (!resSrcTable.Columns.ContainsKey(trackedCol))
                {
                    resSrcTable.AddOrphantColumn(trackedCol);
                }
            }

            // returns prepared SourceTable object
            return resSrcTable;
        }

        /// <summary>
        /// Creates array of <see cref="SourceTable"/> objects representing all tables of the tracked database.
        /// </summary>
        /// <param name="initStoreSrcColId">Determines if <see cref="SourceColumn"/> object property 
        /// <see cref="SourceColumn.StoringColumnId"/> will be initialized with storage database source column id.
        /// </param>
        /// <returns>The array of <see cref="SourceTable"/> objects representing all source database tables.</returns>
        public SourceTable[] CreateAllSourceTables(bool initStoreSrcColId)
        {
            // checks if factory initialized
            if (!_isInitialized)
            {
                throw new CdcSchemaInstallerException("The SourceSchemaFactory not initialized, call" +
                    " InitializeFactory method first.");
            }

            int ESTIMATED_TABLES_NUM = _srcTableCol.Rows.Count / 8; // 8 avg number of columns in the table

            // list of processed tables
            List<string> tableNames = new List<string>(ESTIMATED_TABLES_NUM);

            // list of SourceTable objects prepared for result return
            List<SourceTable> resultSrcTables = new List<SourceTable>(ESTIMATED_TABLES_NUM);

            // _srcTableCol contains all user tables and its rows it the source database
            // row per table column, so duplicated table names have to be handled
            foreach (DataRow row in _srcTableCol.Rows)
            {
                string tblName = (string)row["TABLE_NAME"];
                string tblSchema = (string)row["TABLE_SCHEMA"];
                string tblId = string.Format("{0}{1}{2}", tblSchema, CommonHelper.SCHEMA_TABLE_SEPARATOR, tblName);
                // process only on first occurrence
                if (!tableNames.Contains(tblId))
                {
                    tableNames.Add(tblId);
                    resultSrcTables.Add(CreateSourceTable(tblSchema, tblName, initStoreSrcColId));
                }
            }

            return resultSrcTables.ToArray();
        }
    }
}
